package com.aiguigu.cn.marketanalysis

import java.sql.Timestamp

import org.apache.flink.api.common.functions.AggregateFunction
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

/**
  * @author: yangShen
  * @Description: app 不分渠道 进行市场分析
  * @Date: 2020/5/6 15:52 
  */
object AppMarketing {
  def main(args: Array[String]): Unit = {
    val environment = StreamExecutionEnvironment.getExecutionEnvironment
    environment.setParallelism(1)
    environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val dataStream = environment.addSource(new SimulatedEventSource())
      //指定时间戳
      .assignAscendingTimestamps(_.timestamp)
      .filter(_.behavior != "UNINSTALL")
      .map(data =>{
        ( "dummyKey", 1L )
      })
      //以渠道和行为类型分组
      //开起窗口，滑动窗口
      .keyBy(_._1)
      .timeWindow(Time.hours(1),Time.seconds(10))
      //增量聚合，第一参数类是第二个参数的入参
      .aggregate(new CountAgg(), new MarketingCountTotal())

    dataStream.print()

    environment.execute("app marketing job")
  }
}
//自定义计数，预聚和
class CountAgg extends AggregateFunction[(String, Long), Long, Long]{
  override def createAccumulator(): Long = 0L

  override def add(value: (String, Long), accumulator: Long): Long = accumulator + 1

  override def getResult(accumulator: Long): Long = accumulator

  override def merge(a: Long, b: Long): Long = a + b
}

//自定义窗口函数，输出MarketingViewCount
class MarketingCountTotal extends WindowFunction[Long, MarketingViewCount, String, TimeWindow]{
  override def apply(key: String, window: TimeWindow, input: Iterable[Long], out: Collector[MarketingViewCount]): Unit = {
    val startTs = new Timestamp(window.getStart).toString
    val endTs = new Timestamp(window.getEnd).toString
    //此处应该循环elements进行去重： 1.遍历一边塞到set中，2.数据大的话要用布隆过滤
    val count = input.iterator.next()
    out.collect(MarketingViewCount(startTs, endTs, "app marketing", "total", count))
  }
}
